package com.hudson.store.dispatch;

import com.hudson.store.concurrent.ThreadPoolConstant;
import com.hudson.store.index.IndexService;
import com.hudson.store.queue.TopicQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

public class CommitLogDispatchService {
    private static final Logger logger = LoggerFactory.getLogger(CommitLogDispatchService.class);

    private static final List<CommitLogDispatchExec> dispatchExecList;
    private static final AtomicBoolean isStart = new AtomicBoolean(false);
    private static final AtomicBoolean isRunning = new AtomicBoolean(false);


    static {
        dispatchExecList = new CopyOnWriteArrayList<>();
        dispatchExecList.add(new IndexDispatch());
        dispatchExecList.add(new ConsumeQueueDispatch());

    }

    private static final BlockingQueue<DispatchDto> dispatchDtoBlockingQueue = new LinkedBlockingQueue<>(2_0000);

    static Runnable runnable = () -> {
        isRunning.set(true);
        try {
            while (isStart.get()) {
                try {
                    DispatchDto dispatchDto = dispatchDtoBlockingQueue.take();
                    dispatchExecList.forEach(exec -> {
                        try {
                            exec.exec(dispatchDto);
                        } catch (Exception e) {
                            logger.error("Error executing dispatch task: {}", dispatchDto, e);
                        }
                    });
                } catch (InterruptedException e) {
                    logger.info("Dispatch thread interrupted");
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e) {
                    logger.error("Error in dispatch loop", e);
                }
            }
        } finally {
            isRunning.set(false);
            isStart.set(false);
        }
    };

    public static void start() {
        if (isStart.compareAndSet(false, true)) {

            ThreadPoolConstant.executorService.submit(runnable);
            logger.info("CommitLogDispatchService started");
        }
    }


    public static boolean isRunning() {
        return isRunning.get();
    }

    static class IndexDispatch implements CommitLogDispatchExec {

        @Override
        public void exec(DispatchDto dispatchDto) {
            try {
                IndexService.putIndex(dispatchDto);
            } catch (Exception e) {
                logger.error("Error in IndexDispatch for dispatchDto: {}", dispatchDto, e);
                throw e;
            }
        }
    }

    static class ConsumeQueueDispatch implements CommitLogDispatchExec {

        @Override
        public void exec(DispatchDto dispatchDto) {
            try {
                TopicQueue.write(
                        dispatchDto.getTopic(),
                        dispatchDto.getQueueId(),
                        dispatchDto.getCommitOffset(),
                        dispatchDto.getMessageSize(),
                        dispatchDto.getTag()
                );
            } catch (Exception e) {
                logger.error("Error in ConsumeQueueDispatch for dispatchDto: {}", dispatchDto, e);
                throw e;
            }
        }
    }

    public static boolean putDispatchDto(DispatchDto dispatchDto) {
        if (!isStart.get()) {
            logger.warn("CommitLogDispatchService is not started, dropping dispatchDto: {}", dispatchDto);
            return false;
        }
        return dispatchDtoBlockingQueue.offer(dispatchDto);
    }

    public static int getQueueSize() {
        return dispatchDtoBlockingQueue.size();
    }
}
